package com.graduate.emos.wx.task;

import com.graduate.emos.wx.db.pojo.MessageEntity;
import com.graduate.emos.wx.db.pojo.MessageRefEntity;
import com.graduate.emos.wx.exception.EmosException;
import com.graduate.emos.wx.service.MessageService;
import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

@Component
@Slf4j
public class MessageTask {
    @Autowired
    private ConnectionFactory factory;

    @Autowired
    private MessageService messageService;

    /*消息同步发送到消息队列*/
    public void send(String topic, MessageEntity entity) {
        /*保存消息数据的MongoDB集合中*/
        String id = messageService.insertMessage(entity);
        try (Connection connection = factory.newConnection();
                /*创建消息通道Channel*/
             Channel channel = connection.createChannel();
        ) {
            /*连接队列*/
            channel.queueDeclare(topic, true, false, false, null);
            /*-*/
            HashMap map = new HashMap();
            map.put("messageId", id);
            AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder().headers(map).build();
            /*发送消息*/
            channel.basicPublish("", topic, properties, entity.getMsg().getBytes(StandardCharsets.UTF_8));
            log.debug("消息发送成功");
        } catch (Exception e) {
            log.debug("执行异常", e);
            throw new EmosException("向MQ发送消息失败！");
        }
    }

    /*消息异步发送到消息队列*/
    @Async
    public void sendAsync(String topic, MessageEntity entity) {
        send(topic, entity);
    }

    /*从消息队列接收消息(同步)*/
    public int receive(String topic) {
        int messageCount = 0;
        /*建立连接创建通道*/
        try (Connection connection = factory.newConnection();
                /*创建消息通道Channel*/
             Channel channel = connection.createChannel();
        ) {
            /*连接队列*/
            channel.queueDeclare(topic, true, false, false, null);
            /*循环从消息队列中读取数据*/
            while (true) {
                GetResponse response = channel.basicGet(topic, false);
                if (!Objects.isNull(response)) {
                    /*提取要绑定的数据*/
                    AMQP.BasicProperties props = response.getProps();
                    /*获取请求头的数据*/
                    Map<String, Object> map = props.getHeaders();
                    /*获取messageId*/
                    String messageId = map.get("messageId").toString();
                    /*获取消息正文*/
                    byte[] body = response.getBody();
                    String message = new String(body);
                    log.debug("从RabbitMQ中接收消息" + message);
                    MessageRefEntity entity = new MessageRefEntity();
                    entity.setMessageId(messageId);
                    entity.setReceiverId(Integer.parseInt(topic));
                    entity.setReadFlag(false);
                    entity.setLastFlag(true);
                    /*将消息数据写到MongoDB数据库*/
                    messageService.insertMessageRef(entity);
                    /*返回一个Ack应答*/
                    long deliveryTag = response.getEnvelope().getDeliveryTag();
                    channel.basicAck(deliveryTag, false);
                    messageCount++;
                } else {
                    break;
                }
            }

        } catch (Exception e) {
            log.debug("执行异常", e);
        }
        return messageCount;
    }

    /*从消息队列接收消息(异步)*/
    @Async
    public int receiveAsync(String topic) {
        return receive(topic);
    }

    /*删除消息队列(同步)*/
    public void deleteQueue(String topic) {
        try (Connection connection = factory.newConnection();
                /*创建消息通道Channel*/
             final Channel channel = connection.createChannel();
        ) {
            /*连接队列*/
            channel.queueDelete(topic);
            log.debug("消息队列删除成功！");
        } catch (Exception e) {
            log.debug("执行异常", e);
            throw new EmosException("删除消息队列失败！");
        }
    }

    /*删除消息队列(异步)*/
    @Async
    public void deleteQueueAsync(String topic) {
        deleteQueue(topic);
    }
}